package cn.jiyun.kafka.stream.listener;

import cn.jiyun.kafka.stream.entity.WordCount;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Arrays;
import java.util.Date;

/**
 *  kafka stream流处理器类
 */
@EnableBinding({WordCountListener.WordCountProcessor.class})
public class WordCountListener {

    //流处理方案
    @StreamListener("word-count-stream-topic")
    @SendTo("word-count-stream-consumer")
    public KStream<String, WordCount> process(KStream<String,String> input){
        return input.flatMapValues(value-> Arrays.asList(value.split(" "))).
                map((key,value)->new KeyValue<>(value,value)).groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(15)))
                .count(Materialized.as("word-counts-store"))
                .toStream()
                .map((key,value)->new KeyValue<>(null,
                        new WordCount(key.key(),value,new Date(key.window().start()),
                                new Date(key.window().end()))));
    }


    public interface WordCountProcessor{
        // 指定输入topic
        @Input("word-count-stream-topic")
        KStream<String,String> input();
        // 指定输出topic
        @Output("word-count-stream-consumer")
        KStream<String, WordCount> output();
    }
}
